package hbasecli.hbase;

import cn.hutool.json.JSONUtil;
import lombok.Getter;
import lombok.Setter;
import org.apache.hadoop.hbase.util.Bytes;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Getter
@Setter
public class HbaseRecord {
    private byte[] rowKey;
    private List<HbaseRecordColumn> columns = new ArrayList<>();

    public HbaseRecordOutput toOutput() {
        HbaseRecordOutput output = new HbaseRecordOutput();
        List<Byte> rowKeyAsList = new ArrayList<>(this.rowKey.length);
        for (byte b : rowKey) {
            rowKeyAsList.add(b);
        }
        output.setRowKey(rowKeyAsList);
        Map<String, List<Byte>> map = new HashMap<>();
        Map<String, String> hint = new HashMap<>();  // 给前端的提示信息
        hint.put("rowKey:boolean", safeBytes2BooleanStr(rowKey));
        hint.put("rowKey:integer", safeBytes2IntegerStr(rowKey));
        hint.put("rowKey:short", safeBytes2ShortStr(rowKey));
        hint.put("rowKey:string", safeBytes2Str(rowKey));
        hint.put("rowKey:long", safeBytes2LongStr(rowKey));
        hint.put("rowKey:double", safeBytes2DoubleStr(rowKey));
        hint.put("rowKey:float", safeBytes2FloatStr(rowKey));
        for (HbaseRecordColumn column : columns) {
            List<Byte> valueAsList = new ArrayList<>(column.value.length);
            for (byte b : column.value) {
                valueAsList.add(b);
            }
            String familyQualifier = column.family + ":" + column.qualifier;
            map.put(familyQualifier, valueAsList);
            // 前端做不了这些数据类型转换
            hint.put(familyQualifier + ":boolean", safeBytes2BooleanStr(column.value));
            hint.put(familyQualifier + ":integer", safeBytes2IntegerStr(column.value));
            hint.put(familyQualifier + ":short", safeBytes2ShortStr(column.value));
            hint.put(familyQualifier + ":string", safeBytes2Str(column.value));
            hint.put(familyQualifier + ":long", safeBytes2LongStr(column.value));
            hint.put(familyQualifier + ":double", safeBytes2DoubleStr(column.value));
            hint.put(familyQualifier + ":float", safeBytes2FloatStr(column.value));
        }
        output.setCol2val(map);
        output.setHint(hint);
        return output;
    }


    public String safeBytes2Str(byte[] bytes) {
        try {
            return Bytes.toString(bytes);
        } catch (Exception e) {
            return "[ERROR STRING]";
        }
    }

    public String safeBytes2ShortStr(byte[] bytes) {
        try {
            return Short.toString(Bytes.toShort(bytes));
        } catch (Exception e) {
            return "[ERROR SHORT]";
        }
    }

    public String safeBytes2IntegerStr(byte[] bytes) {
        try {
            return Integer.toString(Bytes.toInt(bytes));
        } catch (Exception e) {
            return "[ERROR INTEGER]";
        }
    }

    public String safeBytes2BooleanStr(byte[] bytes) {
        try {
            return Boolean.toString(Bytes.toBoolean(bytes));
        } catch (Exception e) {
            return "[ERROR BOOLEAN]";
        }
    }

    public String safeBytes2LongStr(byte[] bytes) {
        try {
            return Long.toString(Bytes.toLong(bytes));
        } catch (Exception e) {
            return "[ERROR LONG]";
        }
    }

    public String safeBytes2FloatStr(byte[] bytes) {
        try {
            return Float.toString(Bytes.toFloat(bytes));
        } catch (Exception e) {
            return "[ERROR FLOAT]";
        }
    }

    public String safeBytes2DoubleStr(byte[] bytes) {
        try {
            return Double.toString(Bytes.toDouble(bytes));
        } catch (Exception e) {
            return "[ERROR DOUBLE]";
        }
    }

    @Override
    public String toString() {
        return "HbaseRecord:" + JSONUtil.toJsonStr(this);
    }

    @Getter
    @Setter
    public static class HbaseRecordColumn {
        private String family;
        private String qualifier;
        private String dataType;
        private byte[] value;
    }
}